--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 11a9ea989f491936c6e1abc36c8685232356f84c
Parents : 1ac9c3a
Author : Mark Qvist <mark@unsigned.io>
Date : 2021-08-28T15:57:22+02:00
Fixed indentation.
Changes
Diff
diff --git a/LXMF/LXMF.py b/LXMF/LXMF.py
index cfe0d97..3612902 100644
--- a/LXMF/LXMF.py
+++ b/LXMF/LXMF.py
@@ -7,460 +7,460 @@ import RNS.vendor.umsgpack as msgpack
APP_NAME = "lxmf"
class LXMessage:
- DRAFT = 0x00
- OUTBOUND = 0x01
- SENDING = 0x02
- SENT = 0x04
- DELIVERED = 0x08
- FAILED = 0xFF
- states = [DRAFT, OUTBOUND, SENDING, SENT, DELIVERED, FAILED]
-
- UNKNOWN = 0x00
- PACKET = 0x01
- RESOURCE = 0x02
- representations = [UNKNOWN, PACKET, RESOURCE]
-
- OPPORTUNISTIC = 0x01
- DIRECT = 0x02
- PROPAGATED = 0x03
- valid_methods = [OPPORTUNISTIC, DIRECT, PROPAGATED]
-
- SOURCE_UNKNOWN = 0x01
- SIGNATURE_INVALID = 0x02
- unverified_reasons = [SOURCE_UNKNOWN, SIGNATURE_INVALID]
-
- DESTINATION_LENGTH = RNS.Identity.TRUNCATED_HASHLENGTH//8
- SIGNATURE_LENGTH = RNS.Identity.SIGLENGTH//8
-
- # LXMF overhead is 99 bytes per message:
- # 10 bytes for destination hash
- # 10 bytes for source hash
- # 64 bytes for Ed25519 signature
- # 8 bytes for timestamp
- # 7 bytes for msgpack structure
- LXMF_OVERHEAD = 2*DESTINATION_LENGTH + SIGNATURE_LENGTH + 8 + 7
-
- # With an MTU of 500, the maximum amount of data
- # we can send in a single encrypted packet is
- # 383 bytes.
- ENCRYPTED_PACKET_MDU = RNS.Packet.ENCRYPTED_MDU
-
- # The max content length we can fit in LXMF message
- # inside a single RNS packet is the encrypted MDU, minus
- # the LXMF overhead. We can optimise a bit though, by
- # inferring the destination hash from the destination
- # field of the packet, therefore we also add the length
- # of a destination hash to the calculation. With default
- # RNS and LXMF parameters, the largest single-packet
- # LXMF message we can send is 294 bytes. If a message
- # is larger than that, a Reticulum link will be used.
- ENCRYPTED_PACKET_MAX_CONTENT = ENCRYPTED_PACKET_MDU - LXMF_OVERHEAD + DESTINATION_LENGTH
-
- # Links can carry a larger MDU, due to less overhead per
- # packet. The link MDU with default Reticulum parameters
- # is 415 bytes.
- LINK_PACKET_MDU = RNS.Link.MDU
-
- # Which means that we can deliver single-packet LXMF
- # messages with content of up to 316 bytes over a link.
- # If a message is larger than that, LXMF will sequence
- # and transfer it as a RNS resource over the link instead.
- LINK_PACKET_MAX_CONTENT = LINK_PACKET_MDU - LXMF_OVERHEAD
-
- # For plain packets without encryption, we can
- # fit up to 324 bytes of content.
- PLAIN_PACKET_MDU = RNS.Packet.PLAIN_MDU
- PLAIN_PACKET_MAX_CONTENT = PLAIN_PACKET_MDU - LXMF_OVERHEAD + DESTINATION_LENGTH
-
- # Descriptive strings regarding transport encryption
- ENCRYPTION_DESCRIPTION_AES = "AES-128"
- ENCRYPTION_DESCRIPTION_EC = "Curve25519"
- ENCRYPTION_DESCRIPTION_UNENCRYPTED = "Unencrypted"
-
- def __str__(self):
- if self.hash != None:
- return "<LXMessage "+RNS.hexrep(self.hash, delimit=False)+">"
- else:
- return "<LXMessage>"
-
- def __init__(self, destination, source, content = "", title = "", fields = None, desired_method = None, destination_hash = None, source_hash = None):
- if isinstance(destination, RNS.Destination) or destination == None:
- self.__destination = destination
- if destination != None:
- self.destination_hash = destination.hash
- else:
- self.destination_hash = destination_hash
- else:
- raise ValueError("LXMessage initialised with invalid destination")
-
- if isinstance(source, RNS.Destination) or source == None:
- self.__source = source
- if source != None:
- self.source_hash = source.hash
- else:
- self.source_hash = source_hash
- else:
- raise ValueError("LXMessage initialised with invalid source")
-
- self.set_title_from_string(title)
- self.set_content_from_string(content)
-
- self.set_fields(fields)
-
- self.payload = None
- self.timestamp = None
- self.signature = None
- self.hash = None
- self.packed = None
- self.progress = None
- self.state = LXMessage.DRAFT
- self.method = LXMessage.UNKNOWN
-
- self.incoming = False
- self.signature_validated = False
- self.unverified_reason = None
-
- self.representation = LXMessage.UNKNOWN
- self.desired_method = desired_method
- self.delivery_attempts = 0
- self.transport_encrypted = False
- self.transport_encryption = None
- self.packet_representation = None
- self.resource_representation = None
- self.__delivery_destination = None
- self.__delivery_callback = None
- self.failed_callback = None
-
- def set_title_from_string(self, title_string):
- self.title = title_string.encode("utf-8")
-
- def set_title_from_bytes(self, title_bytes):
- self.title = title_bytes
-
- def title_as_string(self):
- return self.title.decode("utf-8")
-
- def set_content_from_string(self, content_string):
- self.content = content_string.encode("utf-8")
-
- def set_content_from_bytes(self, content_bytes):
- self.content = content_bytes
-
- def content_as_string(self):
- return self.content.decode("utf-8")
-
- def set_fields(self, fields):
- if isinstance(fields, dict) or fields == None:
- self.fields = fields
- else:
- raise ValueError("LXMessage property \"fields\" can only be dict or None")
-
- def get_fields(self):
- return self.__fields
-
- def set_destination(self, destination):
- if self.destination == None:
- if isinstance(destination, RNS.Destination):
- self.__destination = destination
- else:
- raise ValueError("Invalid destination set on LXMessage")
- else:
- raise ValueError("Cannot reassign destination on LXMessage")
-
- def get_destination(self):
- return self.__destination
-
- def set_source(self, source):
- if self.source == None:
- if isinstance(source, RNS.Destination):
- self.__source = source
- else:
- raise ValueError("Invalid source set on LXMessage")
- else:
- raise ValueError("Cannot reassign source on LXMessage")
-
- def get_source(self):
- return self.__source
-
- def set_delivery_destination(self, delivery_destination):
- self.__delivery_destination = delivery_destination
-
- def register_delivery_callback(self, callback):
- self.__delivery_callback = callback
-
- def register_failed_callback(self, callback):
- self.failed_callback = callback
-
- def pack(self):
- if not self.packed:
- if self.timestamp == None:
- self.timestamp = time.time()
-
- self.payload = [self.timestamp, self.title, self.content, self.fields]
-
- hashed_part = b""
- hashed_part += self.__destination.hash
- hashed_part += self.__source.hash
- hashed_part += msgpack.packb(self.payload)
- self.hash = RNS.Identity.full_hash(hashed_part)
- self.message_id = self.hash
-
- signed_part = b""
- signed_part += hashed_part
- signed_part += self.hash
- self.signature = self.__source.sign(signed_part)
- self.signature_validated = True
-
- self.packed = b""
- self.packed += self.__destination.hash
- self.packed += self.__source.hash
- self.packed += self.signature
- packed_payload = msgpack.packb(self.payload)
- self.packed += packed_payload
- self.packed_size = len(self.packed)
- content_size = len(packed_payload)
-
- # If no desired delivery method has been defined,
- # one will be chosen according to these rules:
- if self.desired_method == None:
- self.desired_method == LXMessage.DIRECT
- # TODO: Expand rules to something more intelligent
-
- if self.desired_method == LXMessage.OPPORTUNISTIC:
- if self.__destination.type == RNS.Destination.SINGLE:
- single_packet_content_limit = LXMessage.ENCRYPTED_PACKET_MAX_CONTENT
- elif self.__destination.type == RNS.Destination.PLAIN:
- single_packet_content_limit = LXMessage.PLAIN_PACKET_MAX_CONTENT
-
- if content_size > single_packet_content_limit:
- raise TypeError("LXMessage desired opportunistic delivery method, but content exceeds single-packet size.")
- else:
- self.method = LXMessage.OPPORTUNISTIC
- self.representation = LXMessage.PACKET
- self.__delivery_destination = self.__destination
-
- elif self.desired_method == LXMessage.DIRECT or self.desired_method == LXMessage.PROPAGATED:
- single_packet_content_limit = LXMessage.LINK_PACKET_MAX_CONTENT
- if content_size <= single_packet_content_limit:
- self.method = self.desired_method
- self.representation = LXMessage.PACKET
- else:
- self.method = self.desired_method
- self.representation = LXMessage.RESOURCE
- else:
- raise ValueError("Attempt to re-pack LXMessage "+str(self)+" that was already packed")
-
- def send(self):
- self.determine_transport_encryption()
-
- if self.method == LXMessage.OPPORTUNISTIC:
- self.__as_packet().send().set_delivery_callback(self.__mark_delivered)
- self.state = LXMessage.SENT
-
- elif self.method == LXMessage.DIRECT:
- self.state = LXMessage.SENDING
-
- if self.representation == LXMessage.PACKET:
- receipt = self.__as_packet().send()
- receipt.set_delivery_callback(self.__mark_delivered)
- receipt.set_timeout_callback(self.__link_packet_timed_out)
-
- elif self.representation == LXMessage.RESOURCE:
- self.resource_representation = self.__as_resource()
-
- elif self.method == LXMessage.PROPAGATED:
- # TODO: Implement propagation
- pass
-
- def determine_transport_encryption(self):
- if RNS.Reticulum.should_allow_unencrypted():
- self.transport_encrypted = False
- self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED
- else:
- if self.method == LXMessage.OPPORTUNISTIC:
- if self.destination.type == RNS.Destination.SINGLE:
- self.transport_encrypted = True
- self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC
- elif destination_type == RNS.Destination.GROUP:
- self.transport_encrypted = True
- self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES
- else:
- self.transport_encrypted = False
- self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED
- elif self.method == LXMessage.DIRECT:
- self.transport_encrypted = True
- self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC
- elif self.method == LXMessage.PROPAGATED:
- if self.destination.type == RNS.Destination.SINGLE:
- self.transport_encrypted = True
- self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC
- elif destination_type == RNS.Destination.GROUP:
- self.transport_encrypted = True
- self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES
- else:
- self.transport_encrypted = False
- self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED
- else:
- self.transport_encrypted = False
- self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED
-
- def __mark_delivered(self, receipt = None):
- RNS.log("Received delivery notification for "+str(self), RNS.LOG_DEBUG)
- self.state = LXMessage.DELIVERED
-
- if self.__delivery_callback != None:
- self.__delivery_callback(self)
-
- def __resource_concluded(self, resource):
- if resource.status == RNS.Resource.COMPLETE:
- self.__mark_delivered()
- else:
- resource.link.teardown()
- self.state = LXMessage.OUTBOUND
-
- def __link_packet_timed_out(self, packet_receipt):
- packet_receipt.destination.teardown()
- self.state = LXMessage.OUTBOUND
-
-
- def __update_transfer_progress(self, resource):
- self.progress = resource.progress()
-
- def __as_packet(self):
- if not self.packed:
- self.pack()
-
- if not self.__delivery_destination:
- raise ValueError("Can't synthesize packet for LXMF message before delivery destination is known")
-
- if self.method == LXMessage.OPPORTUNISTIC:
- return RNS.Packet(self.__delivery_destination, self.packed[LXMessage.DESTINATION_LENGTH:])
- elif self.method == LXMessage.DIRECT or self.method == LXMessage.PROPAGATED:
- return RNS.Packet(self.__delivery_destination, self.packed)
-
- def __as_resource(self):
- if not self.packed:
- self.pack()
-
- if not self.__delivery_destination:
- raise ValueError("Can't synthesize resource for LXMF message before delivery destination is known")
-
- if not self.__delivery_destination.type == RNS.Destination.LINK:
- raise TypeError("Tried to synthesize resource for LXMF message on a delivery destination that was not a link")
-
- if not self.__delivery_destination.status == RNS.Link.ACTIVE:
- raise ConnectionError("Tried to synthesize resource for LXMF message on a link that was not active")
-
- self.progress = 0.0
- return RNS.Resource(self.packed, self.__delivery_destination, callback = self.__resource_concluded, progress_callback = self.__update_transfer_progress)
-
- def write_to_directory(self, directory_path):
- file_name = RNS.hexrep(self.hash, delimit=False)
- file_path = directory_path+"/"+file_name
-
- try:
- if not self.packed:
- self.pack()
-
- container = {
- "state": self.state,
- "lxmf_bytes": self.packed,
- "transport_encrypted": self.transport_encrypted,
- "transport_encryption": self.transport_encryption
- }
-
- packed_container = msgpack.packb(container)
-
- file = open(file_path, "wb")
- file.write(packed_container)
- file.close()
-
- return file_path
-
- except Exception as e:
- RNS.log("Error while writing LXMF message to file \""+str(file_path)+"\". The contained exception was: "+str(e), RNS.LOG_ERROR)
- return None
-
-
- @staticmethod
- def unpack_from_bytes(lxmf_bytes):
- destination_hash = lxmf_bytes[:LXMessage.DESTINATION_LENGTH]
- source_hash = lxmf_bytes[LXMessage.DESTINATION_LENGTH:2*LXMessage.DESTINATION_LENGTH]
- signature = lxmf_bytes[2*LXMessage.DESTINATION_LENGTH:2*LXMessage.DESTINATION_LENGTH+LXMessage.SIGNATURE_LENGTH]
- packed_payload = lxmf_bytes[2*LXMessage.DESTINATION_LENGTH+LXMessage.SIGNATURE_LENGTH:]
- hashed_part = b"" + destination_hash + source_hash + packed_payload
- message_hash = RNS.Identity.full_hash(hashed_part)
- signed_part = b"" + hashed_part + message_hash
- unpacked_payload = msgpack.unpackb(packed_payload)
- timestamp = unpacked_payload[0]
- title_bytes = unpacked_payload[1]
- content_bytes = unpacked_payload[2]
- fields = unpacked_payload[3]
-
- destination_identity = RNS.Identity.recall(destination_hash)
- if destination_identity != None:
- destination = RNS.Destination(destination_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "delivery")
- else:
- destination = None
-
- source_identity = RNS.Identity.recall(source_hash)
- if source_identity != None:
- source = RNS.Destination(source_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "delivery")
- else:
- source = None
-
- message = LXMessage(
- destination = destination,
- source = source,
- content = "",
- title = "",
- fields = fields,
- destination_hash = destination_hash,
- source_hash = source_hash)
-
- message.hash = message_hash
- message.signature = signature
- message.incoming = True
- message.timestamp = timestamp
- message.packed = lxmf_bytes
- message.packed_size = len(lxmf_bytes)
- message.set_title_from_bytes(title_bytes)
- message.set_content_from_bytes(content_bytes)
-
- try:
- if source:
- if source.identity.validate(signature, signed_part):
- message.signature_validated = True
- else:
- message.signature_validated = False
- message.unverified_reason = LXMessage.SIGNATURE_INVALID
- else:
- signature_validated = False
- message.unverified_reason = LXMessage.SOURCE_UNKNOWN
- RNS.log("Unpacked LXMF message signature could not be validated, since source identity is unknown", RNS.LOG_DEBUG)
- except Exception as e:
- message.signature_validated = False
- RNS.log("Error while validating LXMF message signature. The contained exception was: "+str(e), RNS.LOG_ERROR)
-
- return message
-
- @staticmethod
- def unpack_from_file(lxmf_file_handle):
- try:
- container = msgpack.unpackb(lxmf_file_handle.read())
- lxm = LXMessage.unpack_from_bytes(container["lxmf_bytes"])
-
- if "state" in container:
- lxm.state = container["state"]
- if "transport_encrypted" in container:
- lxm.transport_encrypted = container["transport_encrypted"]
- if "transport_encryption" in container:
- lxm.transport_encryption = container["transport_encryption"]
-
- return lxm
- except Exception as e:
- RNS.log("Could not unpack LXMessage from file. The contained exception was: "+str(e), RNS.LOG_ERROR)
- return None
+ DRAFT = 0x00
+ OUTBOUND = 0x01
+ SENDING = 0x02
+ SENT = 0x04
+ DELIVERED = 0x08
+ FAILED = 0xFF
+ states = [DRAFT, OUTBOUND, SENDING, SENT, DELIVERED, FAILED]
+
+ UNKNOWN = 0x00
+ PACKET = 0x01
+ RESOURCE = 0x02
+ representations = [UNKNOWN, PACKET, RESOURCE]
+
+ OPPORTUNISTIC = 0x01
+ DIRECT = 0x02
+ PROPAGATED = 0x03
+ valid_methods = [OPPORTUNISTIC, DIRECT, PROPAGATED]
+
+ SOURCE_UNKNOWN = 0x01
+ SIGNATURE_INVALID = 0x02
+ unverified_reasons = [SOURCE_UNKNOWN, SIGNATURE_INVALID]
+
+ DESTINATION_LENGTH = RNS.Identity.TRUNCATED_HASHLENGTH//8
+ SIGNATURE_LENGTH = RNS.Identity.SIGLENGTH//8
+
+ # LXMF overhead is 99 bytes per message:
+ # 10 bytes for destination hash
+ # 10 bytes for source hash
+ # 64 bytes for Ed25519 signature
+ # 8 bytes for timestamp
+ # 7 bytes for msgpack structure
+ LXMF_OVERHEAD = 2*DESTINATION_LENGTH + SIGNATURE_LENGTH + 8 + 7
+
+ # With an MTU of 500, the maximum amount of data
+ # we can send in a single encrypted packet is
+ # 383 bytes.
+ ENCRYPTED_PACKET_MDU = RNS.Packet.ENCRYPTED_MDU
+
+ # The max content length we can fit in LXMF message
+ # inside a single RNS packet is the encrypted MDU, minus
+ # the LXMF overhead. We can optimise a bit though, by
+ # inferring the destination hash from the destination
+ # field of the packet, therefore we also add the length
+ # of a destination hash to the calculation. With default
+ # RNS and LXMF parameters, the largest single-packet
+ # LXMF message we can send is 294 bytes. If a message
+ # is larger than that, a Reticulum link will be used.
+ ENCRYPTED_PACKET_MAX_CONTENT = ENCRYPTED_PACKET_MDU - LXMF_OVERHEAD + DESTINATION_LENGTH
+
+ # Links can carry a larger MDU, due to less overhead per
+ # packet. The link MDU with default Reticulum parameters
+ # is 415 bytes.
+ LINK_PACKET_MDU = RNS.Link.MDU
+
+ # Which means that we can deliver single-packet LXMF
+ # messages with content of up to 316 bytes over a link.
+ # If a message is larger than that, LXMF will sequence
+ # and transfer it as a RNS resource over the link instead.
+ LINK_PACKET_MAX_CONTENT = LINK_PACKET_MDU - LXMF_OVERHEAD
+
+ # For plain packets without encryption, we can
+ # fit up to 324 bytes of content.
+ PLAIN_PACKET_MDU = RNS.Packet.PLAIN_MDU
+ PLAIN_PACKET_MAX_CONTENT = PLAIN_PACKET_MDU - LXMF_OVERHEAD + DESTINATION_LENGTH
+
+ # Descriptive strings regarding transport encryption
+ ENCRYPTION_DESCRIPTION_AES = "AES-128"
+ ENCRYPTION_DESCRIPTION_EC = "Curve25519"
+ ENCRYPTION_DESCRIPTION_UNENCRYPTED = "Unencrypted"
+
+ def __str__(self):
+ if self.hash != None:
+ return "<LXMessage "+RNS.hexrep(self.hash, delimit=False)+">"
+ else:
+ return "<LXMessage>"
+
+ def __init__(self, destination, source, content = "", title = "", fields = None, desired_method = None, destination_hash = None, source_hash = None):
+ if isinstance(destination, RNS.Destination) or destination == None:
+ self.__destination = destination
+ if destination != None:
+ self.destination_hash = destination.hash
+ else:
+ self.destination_hash = destination_hash
+ else:
+ raise ValueError("LXMessage initialised with invalid destination")
+
+ if isinstance(source, RNS.Destination) or source == None:
+ self.__source = source
+ if source != None:
+ self.source_hash = source.hash
+ else:
+ self.source_hash = source_hash
+ else:
+ raise ValueError("LXMessage initialised with invalid source")
+
+ self.set_title_from_string(title)
+ self.set_content_from_string(content)
+
+ self.set_fields(fields)
+
+ self.payload = None
+ self.timestamp = None
+ self.signature = None
+ self.hash = None
+ self.packed = None
+ self.progress = None
+ self.state = LXMessage.DRAFT
+ self.method = LXMessage.UNKNOWN
+
+ self.incoming = False
+ self.signature_validated = False
+ self.unverified_reason = None
+
+ self.representation = LXMessage.UNKNOWN
+ self.desired_method = desired_method
+ self.delivery_attempts = 0
+ self.transport_encrypted = False
+ self.transport_encryption = None
+ self.packet_representation = None
+ self.resource_representation = None
+ self.__delivery_destination = None
+ self.__delivery_callback = None
+ self.failed_callback = None
+
+ def set_title_from_string(self, title_string):
+ self.title = title_string.encode("utf-8")
+
+ def set_title_from_bytes(self, title_bytes):
+ self.title = title_bytes
+
+ def title_as_string(self):
+ return self.title.decode("utf-8")
+
+ def set_content_from_string(self, content_string):
+ self.content = content_string.encode("utf-8")
+
+ def set_content_from_bytes(self, content_bytes):
+ self.content = content_bytes
+
+ def content_as_string(self):
+ return self.content.decode("utf-8")
+
+ def set_fields(self, fields):
+ if isinstance(fields, dict) or fields == None:
+ self.fields = fields
+ else:
+ raise ValueError("LXMessage property \"fields\" can only be dict or None")
+
+ def get_fields(self):
+ return self.__fields
+
+ def set_destination(self, destination):
+ if self.destination == None:
+ if isinstance(destination, RNS.Destination):
+ self.__destination = destination
+ else:
+ raise ValueError("Invalid destination set on LXMessage")
+ else:
+ raise ValueError("Cannot reassign destination on LXMessage")
+
+ def get_destination(self):
+ return self.__destination
+
+ def set_source(self, source):
+ if self.source == None:
+ if isinstance(source, RNS.Destination):
+ self.__source = source
+ else:
+ raise ValueError("Invalid source set on LXMessage")
+ else:
+ raise ValueError("Cannot reassign source on LXMessage")
+
+ def get_source(self):
+ return self.__source
+
+ def set_delivery_destination(self, delivery_destination):
+ self.__delivery_destination = delivery_destination
+
+ def register_delivery_callback(self, callback):
+ self.__delivery_callback = callback
+
+ def register_failed_callback(self, callback):
+ self.failed_callback = callback
+
+ def pack(self):
+ if not self.packed:
+ if self.timestamp == None:
+ self.timestamp = time.time()
+
+ self.payload = [self.timestamp, self.title, self.content, self.fields]
+
+ hashed_part = b""
+ hashed_part += self.__destination.hash
+ hashed_part += self.__source.hash
+ hashed_part += msgpack.packb(self.payload)
+ self.hash = RNS.Identity.full_hash(hashed_part)
+ self.message_id = self.hash
+
+ signed_part = b""
+ signed_part += hashed_part
+ signed_part += self.hash
+ self.signature = self.__source.sign(signed_part)
+ self.signature_validated = True
+
+ self.packed = b""
+ self.packed += self.__destination.hash
+ self.packed += self.__source.hash
+ self.packed += self.signature
+ packed_payload = msgpack.packb(self.payload)
+ self.packed += packed_payload
+ self.packed_size = len(self.packed)
+ content_size = len(packed_payload)
+
+ # If no desired delivery method has been defined,
+ # one will be chosen according to these rules:
+ if self.desired_method == None:
+ self.desired_method == LXMessage.DIRECT
+ # TODO: Expand rules to something more intelligent
+
+ if self.desired_method == LXMessage.OPPORTUNISTIC:
+ if self.__destination.type == RNS.Destination.SINGLE:
+ single_packet_content_limit = LXMessage.ENCRYPTED_PACKET_MAX_CONTENT
+ elif self.__destination.type == RNS.Destination.PLAIN:
+ single_packet_content_limit = LXMessage.PLAIN_PACKET_MAX_CONTENT
+
+ if content_size > single_packet_content_limit:
+ raise TypeError("LXMessage desired opportunistic delivery method, but content exceeds single-packet size.")
+ else:
+ self.method = LXMessage.OPPORTUNISTIC
+ self.representation = LXMessage.PACKET
+ self.__delivery_destination = self.__destination
+
+ elif self.desired_method == LXMessage.DIRECT or self.desired_method == LXMessage.PROPAGATED:
+ single_packet_content_limit = LXMessage.LINK_PACKET_MAX_CONTENT
+ if content_size <= single_packet_content_limit:
+ self.method = self.desired_method
+ self.representation = LXMessage.PACKET
+ else:
+ self.method = self.desired_method
+ self.representation = LXMessage.RESOURCE
+ else:
+ raise ValueError("Attempt to re-pack LXMessage "+str(self)+" that was already packed")
+
+ def send(self):
+ self.determine_transport_encryption()
+
+ if self.method == LXMessage.OPPORTUNISTIC:
+ self.__as_packet().send().set_delivery_callback(self.__mark_delivered)
+ self.state = LXMessage.SENT
+
+ elif self.method == LXMessage.DIRECT:
+ self.state = LXMessage.SENDING
+
+ if self.representation == LXMessage.PACKET:
+ receipt = self.__as_packet().send()
+ receipt.set_delivery_callback(self.__mark_delivered)
+ receipt.set_timeout_callback(self.__link_packet_timed_out)
+
+ elif self.representation == LXMessage.RESOURCE:
+ self.resource_representation = self.__as_resource()
+
+ elif self.method == LXMessage.PROPAGATED:
+ # TODO: Implement propagation
+ pass
+
+ def determine_transport_encryption(self):
+ if RNS.Reticulum.should_allow_unencrypted():
+ self.transport_encrypted = False
+ self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED
+ else:
+ if self.method == LXMessage.OPPORTUNISTIC:
+ if self.destination.type == RNS.Destination.SINGLE:
+ self.transport_encrypted = True
+ self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC
+ elif destination_type == RNS.Destination.GROUP:
+ self.transport_encrypted = True
+ self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES
+ else:
+ self.transport_encrypted = False
+ self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED
+ elif self.method == LXMessage.DIRECT:
+ self.transport_encrypted = True
+ self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC
+ elif self.method == LXMessage.PROPAGATED:
+ if self.destination.type == RNS.Destination.SINGLE:
+ self.transport_encrypted = True
+ self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC
+ elif destination_type == RNS.Destination.GROUP:
+ self.transport_encrypted = True
+ self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES
+ else:
+ self.transport_encrypted = False
+ self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED
+ else:
+ self.transport_encrypted = False
+ self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED
+
+ def __mark_delivered(self, receipt = None):
+ RNS.log("Received delivery notification for "+str(self), RNS.LOG_DEBUG)
+ self.state = LXMessage.DELIVERED
+
+ if self.__delivery_callback != None:
+ self.__delivery_callback(self)
+
+ def __resource_concluded(self, resource):
+ if resource.status == RNS.Resource.COMPLETE:
+ self.__mark_delivered()
+ else:
+ resource.link.teardown()
+ self.state = LXMessage.OUTBOUND
+
+ def __link_packet_timed_out(self, packet_receipt):
+ packet_receipt.destination.teardown()
+ self.state = LXMessage.OUTBOUND
+
+
+ def __update_transfer_progress(self, resource):
+ self.progress = resource.progress()
+
+ def __as_packet(self):
+ if not self.packed:
+ self.pack()
+
+ if not self.__delivery_destination:
+ raise ValueError("Can't synthesize packet for LXMF message before delivery destination is known")
+
+ if self.method == LXMessage.OPPORTUNISTIC:
+ return RNS.Packet(self.__delivery_destination, self.packed[LXMessage.DESTINATION_LENGTH:])
+ elif self.method == LXMessage.DIRECT or self.method == LXMessage.PROPAGATED:
+ return RNS.Packet(self.__delivery_destination, self.packed)
+
+ def __as_resource(self):
+ if not self.packed:
+ self.pack()
+
+ if not self.__delivery_destination:
+ raise ValueError("Can't synthesize resource for LXMF message before delivery destination is known")
+
+ if not self.__delivery_destination.type == RNS.Destination.LINK:
+ raise TypeError("Tried to synthesize resource for LXMF message on a delivery destination that was not a link")
+
+ if not self.__delivery_destination.status == RNS.Link.ACTIVE:
+ raise ConnectionError("Tried to synthesize resource for LXMF message on a link that was not active")
+
+ self.progress = 0.0
+ return RNS.Resource(self.packed, self.__delivery_destination, callback = self.__resource_concluded, progress_callback = self.__update_transfer_progress)
+
+ def write_to_directory(self, directory_path):
+ file_name = RNS.hexrep(self.hash, delimit=False)
+ file_path = directory_path+"/"+file_name
+
+ try:
+ if not self.packed:
+ self.pack()
+
+ container = {
+ "state": self.state,
+ "lxmf_bytes": self.packed,
+ "transport_encrypted": self.transport_encrypted,
+ "transport_encryption": self.transport_encryption
+ }
+
+ packed_container = msgpack.packb(container)
+
+ file = open(file_path, "wb")
+ file.write(packed_container)
+ file.close()
+
+ return file_path
+
+ except Exception as e:
+ RNS.log("Error while writing LXMF message to file \""+str(file_path)+"\". The contained exception was: "+str(e), RNS.LOG_ERROR)
+ return None
+
+
+ @staticmethod
+ def unpack_from_bytes(lxmf_bytes):
+ destination_hash = lxmf_bytes[:LXMessage.DESTINATION_LENGTH]
+ source_hash = lxmf_bytes[LXMessage.DESTINATION_LENGTH:2*LXMessage.DESTINATION_LENGTH]
+ signature = lxmf_bytes[2*LXMessage.DESTINATION_LENGTH:2*LXMessage.DESTINATION_LENGTH+LXMessage.SIGNATURE_LENGTH]
+ packed_payload = lxmf_bytes[2*LXMessage.DESTINATION_LENGTH+LXMessage.SIGNATURE_LENGTH:]
+ hashed_part = b"" + destination_hash + source_hash + packed_payload
+ message_hash = RNS.Identity.full_hash(hashed_part)
+ signed_part = b"" + hashed_part + message_hash
+ unpacked_payload = msgpack.unpackb(packed_payload)
+ timestamp = unpacked_payload[0]
+ title_bytes = unpacked_payload[1]
+ content_bytes = unpacked_payload[2]
+ fields = unpacked_payload[3]
+
+ destination_identity = RNS.Identity.recall(destination_hash)
+ if destination_identity != None:
+ destination = RNS.Destination(destination_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "delivery")
+ else:
+ destination = None
+
+ source_identity = RNS.Identity.recall(source_hash)
+ if source_identity != None:
+ source = RNS.Destination(source_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "delivery")
+ else:
+ source = None
+
+ message = LXMessage(
+ destination = destination,
+ source = source,
+ content = "",
+ title = "",
+ fields = fields,
+ destination_hash = destination_hash,
+ source_hash = source_hash)
+
+ message.hash = message_hash
+ message.signature = signature
+ message.incoming = True
+ message.timestamp = timestamp
+ message.packed = lxmf_bytes
+ message.packed_size = len(lxmf_bytes)
+ message.set_title_from_bytes(title_bytes)
+ message.set_content_from_bytes(content_bytes)
+
+ try:
+ if source:
+ if source.identity.validate(signature, signed_part):
+ message.signature_validated = True
+ else:
+ message.signature_validated = False
+ message.unverified_reason = LXMessage.SIGNATURE_INVALID
+ else:
+ signature_validated = False
+ message.unverified_reason = LXMessage.SOURCE_UNKNOWN
+ RNS.log("Unpacked LXMF message signature could not be validated, since source identity is unknown", RNS.LOG_DEBUG)
+ except Exception as e:
+ message.signature_validated = False
+ RNS.log("Error while validating LXMF message signature. The contained exception was: "+str(e), RNS.LOG_ERROR)
+
+ return message
+
+ @staticmethod
+ def unpack_from_file(lxmf_file_handle):
+ try:
+ container = msgpack.unpackb(lxmf_file_handle.read())
+ lxm = LXMessage.unpack_from_bytes(container["lxmf_bytes"])
+
+ if "state" in container:
+ lxm.state = container["state"]
+ if "transport_encrypted" in container:
+ lxm.transport_encrypted = container["transport_encrypted"]
+ if "transport_encryption" in container:
+ lxm.transport_encryption = container["transport_encryption"]
+
+ return lxm
+ except Exception as e:
+ RNS.log("Could not unpack LXMessage from file. The contained exception was: "+str(e), RNS.LOG_ERROR)
+ return None
class LXMFDeliveryAnnounceHandler:
@@ -470,253 +470,253 @@ class LXMFDeliveryAnnounceHandler:
def received_announce(self, destination_hash, announced_identity, app_data):
for lxmessage in self.lxmrouter.pending_outbound:
- if destination_hash == lxmessage.destination_hash:
- if lxmessage.method == LXMessage.DIRECT:
- lxmessage.next_delivery_attempt = time.time()
+ if destination_hash == lxmessage.destination_hash:
+ if lxmessage.method == LXMessage.DIRECT:
+ lxmessage.next_delivery_attempt = time.time()
- while self.lxmrouter.processing_outbound:
- time.sleep(0.1)
+ while self.lxmrouter.processing_outbound:
+ time.sleep(0.1)
- self.lxmrouter.process_outbound()
+ self.lxmrouter.process_outbound()
class LXMRouter:
- MAX_DELIVERY_ATTEMPTS = 3
- PROCESSING_INTERVAL = 5
- DELIVERY_RETRY_WAIT = 15
- PATH_REQUEST_WAIT = 5
- LINK_MAX_INACTIVITY = 10*60
-
- def __init__(self):
- self.pending_inbound = []
- self.pending_outbound = []
- self.failed_outbound = []
- self.direct_links = {}
- self.delivery_destinations = {}
-
- self.processing_outbound = False
- self.processing_inbound = False
-
- self.identity = RNS.Identity()
- self.lxmf_query_destination = RNS.Destination(None, RNS.Destination.IN, RNS.Destination.PLAIN, APP_NAME, "query")
- self.propagation_destination = RNS.Destination(self.identity, RNS.Destination.IN, RNS.Destination.SINGLE, APP_NAME, "propagation")
-
- RNS.Transport.register_announce_handler(LXMFDeliveryAnnounceHandler(self))
-
- self.__delivery_callback = None
-
- job_thread = threading.Thread(target=self.jobloop)
- job_thread.setDaemon(True)
- job_thread.start()
-
- def register_delivery_identity(self, identity, display_name = None):
- delivery_destination = RNS.Destination(identity, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery")
- delivery_destination.set_packet_callback(self.delivery_packet)
- delivery_destination.set_link_established_callback(self.delivery_link_established)
- delivery_destination.display_name = display_name
-
- if display_name != None:
- delivery_destination.set_default_app_data(display_name.encode("utf-8"))
-
- self.delivery_destinations[delivery_destination.hash] = delivery_destination
- return delivery_destination
-
- def register_delivery_callback(self, callback):
- self.__delivery_callback = callback
-
- def announce(self, destination_hash):
- if destination_hash in self.delivery_destinations:
- delivery_destination = self.delivery_destinations[destination_hash]
- delivery_destination.announce(delivery_destination.display_name.encode("utf-8"))
-
- def handle_outbound(self, lxmessage):
- lxmessage.state = LXMessage.OUTBOUND
- if not lxmessage.packed:
- lxmessage.pack()
-
- lxmessage.determine_transport_encryption()
-
- while self.processing_outbound:
- time.sleep(0.1)
-
- self.pending_outbound.append(lxmessage)
- self.process_outbound()
-
- def lxmf_delivery(self, lxmf_data, destination_type = None):
- try:
- message = LXMessage.unpack_from_bytes(lxmf_data)
-
- if RNS.Reticulum.should_allow_unencrypted():
- message.transport_encrypted = False
- message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED
- else:
- if destination_type == RNS.Destination.SINGLE:
- message.transport_encrypted = True
- message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_RSA
- elif destination_type == RNS.Destination.GROUP:
- message.transport_encrypted = True
- message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES
- elif destination_type == RNS.Destination.LINK:
- message.transport_encrypted = True
- message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC
- else:
- message.transport_encrypted = False
- message.transport_encryption = None
-
- if self.__delivery_callback != None:
- self.__delivery_callback(message)
-
- return True
-
- except Exception as e:
- RNS.log("Could not assemble LXMF message from received data", RNS.LOG_NOTICE)
- RNS.log("The contained exception was: "+str(e), RNS.LOG_DEBUG)
- return False
-
-
- def delivery_packet(self, data, packet):
- try:
- if packet.destination_type != RNS.Destination.LINK:
- lxmf_data = b""
- lxmf_data += packet.destination.hash
- lxmf_data += data
- else:
- lxmf_data = data
-
- if self.lxmf_delivery(lxmf_data, packet.destination_type):
- packet.prove()
-
- except Exception as e:
- RNS.log("Exception occurred while parsing incoming LXMF data.", RNS.LOG_ERROR)
- RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR)
-
- def delivery_link_established(self, link):
- link.set_packet_callback(self.delivery_packet)
- link.set_resource_strategy(RNS.Link.ACCEPT_ALL)
- link.set_resource_started_callback(self.resource_transfer_began)
- link.set_resource_concluded_callback(self.resource_transfer_concluded)
-
- def delivery_link_closed(self, link):
- pass
-
- def resource_transfer_began(self, resource):
- RNS.log("Transfer began for resource "+str(resource), RNS.LOG_DEBUG)
-
- def resource_transfer_concluded(self, resource):
- RNS.log("Transfer concluded for resource "+str(resource), RNS.LOG_DEBUG)
- if resource.status == RNS.Resource.COMPLETE:
- self.lxmf_delivery(resource.data.read(), resource.link.type)
-
- def jobloop(self):
- while (True):
- # TODO: Improve this to scheduling, so manual
- # triggers can delay next run
- self.jobs()
- time.sleep(LXMRouter.PROCESSING_INTERVAL)
-
- def jobs(self):
- self.process_outbound()
- self.clean_links()
-
- def clean_links(self):
- closed_links = []
- for link_hash in self.direct_links:
- link = self.direct_links[link_hash]
- inactive_time = link.inactive_for()
-
- if inactive_time > LXMRouter.LINK_MAX_INACTIVITY:
- link.teardown()
- closed_links.append(link_hash)
-
- for link_hash in closed_links:
- cleaned_link = self.direct_links.pop(link_hash)
- RNS.log("Cleaned link "+str(cleaned_link), RNS.LOG_DEBUG)
-
- def fail_message(self, lxmessage):
- RNS.log(str(lxmessage)+" failed to send", RNS.LOG_DEBUG)
-
- self.pending_outbound.remove(lxmessage)
- self.failed_outbound.append(lxmessage)
-
- lxmessage.state = LXMessage.FAILED
- if lxmessage.failed_callback != None:
- lxmessage.failed_callback(lxmessage)
-
- def process_outbound(self, sender = None):
- if self.processing_outbound:
- return
-
- for lxmessage in self.pending_outbound:
- if lxmessage.state == LXMessage.DELIVERED:
- RNS.log("Delivery has occurred for "+str(lxmessage)+", removing from outbound queue", RNS.LOG_DEBUG)
- self.pending_outbound.remove(lxmessage)
- else:
- RNS.log("Starting outbound processing for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
- # Outbound handling for opportunistic messages
- if lxmessage.method == LXMessage.OPPORTUNISTIC:
- if lxmessage.delivery_attempts <= LXMRouter.MAX_DELIVERY_ATTEMPTS:
- if not hasattr(lxmessage, "next_delivery_attempt") or time.time() > lxmessage.next_delivery_attempt:
- lxmessage.delivery_attempts += 1
- lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT
- RNS.log("Opportunistic delivery attempt "+str(lxmessage.delivery_attempts)+" for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
- lxmessage.send()
- else:
- RNS.log("Max delivery attempts reached for oppertunistic "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
- self.fail_message(lxmessage)
-
- # Outbound handling for messages transferred
- # over a direct link to the final recipient
- elif lxmessage.method == LXMessage.DIRECT:
- if lxmessage.delivery_attempts <= LXMRouter.MAX_DELIVERY_ATTEMPTS:
- delivery_destination_hash = lxmessage.get_destination().hash
-
- if delivery_destination_hash in self.direct_links:
- # A link already exists, so we'll try to use it
- # to deliver the message
- direct_link = self.direct_links[delivery_destination_hash]
- if direct_link.status == RNS.Link.ACTIVE:
- if lxmessage.state != LXMessage.SENDING:
- RNS.log("Starting transfer of "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
- lxmessage.set_delivery_destination(direct_link)
- lxmessage.send()
- else:
- if lxmessage.representation == LXMessage.RESOURCE:
- RNS.log("The transfer of "+str(lxmessage)+" is in progress ("+str(round(lxmessage.progress*100, 1))+"%)", RNS.LOG_DEBUG)
- else:
- RNS.log("Waiting for proof for "+str(lxmessage)+" sent as link packet", RNS.LOG_DEBUG)
- elif direct_link.status == RNS.Link.CLOSED:
- RNS.log("The link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" was closed", RNS.LOG_DEBUG)
- lxmessage.set_delivery_destination(None)
- self.direct_links.pop(delivery_destination_hash)
- lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT
- else:
- # Simply wait for the link to become
- # active or close
- RNS.log("The link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" is pending, waiting for link to become active", RNS.LOG_DEBUG)
- else:
- # No link exists, so we'll try to establish one, but
- # only if we've never tried before, or the retry wait
- # period has elapsed.
- if not hasattr(lxmessage, "next_delivery_attempt") or time.time() > lxmessage.next_delivery_attempt:
- lxmessage.delivery_attempts += 1
- lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT
-
- if lxmessage.delivery_attempts < LXMRouter.MAX_DELIVERY_ATTEMPTS:
- if RNS.Transport.has_path(lxmessage.get_destination().hash):
- RNS.log("Establishing link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" for delivery attempt "+str(lxmessage.delivery_attempts)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
- delivery_link = RNS.Link(lxmessage.get_destination())
- delivery_link.set_link_established_callback(self.process_outbound)
- self.direct_links[delivery_destination_hash] = delivery_link
- else:
- RNS.log("No path known for delivery attempt "+str(lxmessage.delivery_attempts)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+". Requesting path...", RNS.LOG_DEBUG)
- RNS.Transport.request_path(lxmessage.get_destination().hash)
- lxmessage.next_delivery_attempt = time.time() + LXMRouter.PATH_REQUEST_WAIT
- else:
- RNS.log("Max delivery attempts reached for direct "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
- self.fail_message(lxmessage)
-
- # Outbound handling for messages transported via
- # propagation to a LXMF router network.
- elif lxmessage.method == LXMessage.PROPAGATED:
- RNS.log("Attempting propagated delivery for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
- raise NotImplementedError("LXMF propagation is not implemented yet")
\ No newline at end of file
+ MAX_DELIVERY_ATTEMPTS = 3
+ PROCESSING_INTERVAL = 5
+ DELIVERY_RETRY_WAIT = 15
+ PATH_REQUEST_WAIT = 5
+ LINK_MAX_INACTIVITY = 10*60
+
+ def __init__(self):
+ self.pending_inbound = []
+ self.pending_outbound = []
+ self.failed_outbound = []
+ self.direct_links = {}
+ self.delivery_destinations = {}
+
+ self.processing_outbound = False
+ self.processing_inbound = False
+
+ self.identity = RNS.Identity()
+ self.lxmf_query_destination = RNS.Destination(None, RNS.Destination.IN, RNS.Destination.PLAIN, APP_NAME, "query")
+ self.propagation_destination = RNS.Destination(self.identity, RNS.Destination.IN, RNS.Destination.SINGLE, APP_NAME, "propagation")
+
+ RNS.Transport.register_announce_handler(LXMFDeliveryAnnounceHandler(self))
+
+ self.__delivery_callback = None
+
+ job_thread = threading.Thread(target=self.jobloop)
+ job_thread.setDaemon(True)
+ job_thread.start()
+
+ def register_delivery_identity(self, identity, display_name = None):
+ delivery_destination = RNS.Destination(identity, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery")
+ delivery_destination.set_packet_callback(self.delivery_packet)
+ delivery_destination.set_link_established_callback(self.delivery_link_established)
+ delivery_destination.display_name = display_name
+
+ if display_name != None:
+ delivery_destination.set_default_app_data(display_name.encode("utf-8"))
+
+ self.delivery_destinations[delivery_destination.hash] = delivery_destination
+ return delivery_destination
+
+ def register_delivery_callback(self, callback):
+ self.__delivery_callback = callback
+
+ def announce(self, destination_hash):
+ if destination_hash in self.delivery_destinations:
+ delivery_destination = self.delivery_destinations[destination_hash]
+ delivery_destination.announce(delivery_destination.display_name.encode("utf-8"))
+
+ def handle_outbound(self, lxmessage):
+ lxmessage.state = LXMessage.OUTBOUND
+ if not lxmessage.packed:
+ lxmessage.pack()
+
+ lxmessage.determine_transport_encryption()
+
+ while self.processing_outbound:
+ time.sleep(0.1)
+
+ self.pending_outbound.append(lxmessage)
+ self.process_outbound()
+
+ def lxmf_delivery(self, lxmf_data, destination_type = None):
+ try:
+ message = LXMessage.unpack_from_bytes(lxmf_data)
+
+ if RNS.Reticulum.should_allow_unencrypted():
+ message.transport_encrypted = False
+ message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED
+ else:
+ if destination_type == RNS.Destination.SINGLE:
+ message.transport_encrypted = True
+ message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_RSA
+ elif destination_type == RNS.Destination.GROUP:
+ message.transport_encrypted = True
+ message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES
+ elif destination_type == RNS.Destination.LINK:
+ message.transport_encrypted = True
+ message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC
+ else:
+ message.transport_encrypted = False
+ message.transport_encryption = None
+
+ if self.__delivery_callback != None:
+ self.__delivery_callback(message)
+
+ return True
+
+ except Exception as e:
+ RNS.log("Could not assemble LXMF message from received data", RNS.LOG_NOTICE)
+ RNS.log("The contained exception was: "+str(e), RNS.LOG_DEBUG)
+ return False
+
+
+ def delivery_packet(self, data, packet):
+ try:
+ if packet.destination_type != RNS.Destination.LINK:
+ lxmf_data = b""
+ lxmf_data += packet.destination.hash
+ lxmf_data += data
+ else:
+ lxmf_data = data
+
+ if self.lxmf_delivery(lxmf_data, packet.destination_type):
+ packet.prove()
+
+ except Exception as e:
+ RNS.log("Exception occurred while parsing incoming LXMF data.", RNS.LOG_ERROR)
+ RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR)
+
+ def delivery_link_established(self, link):
+ link.set_packet_callback(self.delivery_packet)
+ link.set_resource_strategy(RNS.Link.ACCEPT_ALL)
+ link.set_resource_started_callback(self.resource_transfer_began)
+ link.set_resource_concluded_callback(self.resource_transfer_concluded)
+
+ def delivery_link_closed(self, link):
+ pass
+
+ def resource_transfer_began(self, resource):
+ RNS.log("Transfer began for resource "+str(resource), RNS.LOG_DEBUG)
+
+ def resource_transfer_concluded(self, resource):
+ RNS.log("Transfer concluded for resource "+str(resource), RNS.LOG_DEBUG)
+ if resource.status == RNS.Resource.COMPLETE:
+ self.lxmf_delivery(resource.data.read(), resource.link.type)
+
+ def jobloop(self):
+ while (True):
+ # TODO: Improve this to scheduling, so manual
+ # triggers can delay next run
+ self.jobs()
+ time.sleep(LXMRouter.PROCESSING_INTERVAL)
+
+ def jobs(self):
+ self.process_outbound()
+ self.clean_links()
+
+ def clean_links(self):
+ closed_links = []
+ for link_hash in self.direct_links:
+ link = self.direct_links[link_hash]
+ inactive_time = link.inactive_for()
+
+ if inactive_time > LXMRouter.LINK_MAX_INACTIVITY:
+ link.teardown()
+ closed_links.append(link_hash)
+
+ for link_hash in closed_links:
+ cleaned_link = self.direct_links.pop(link_hash)
+ RNS.log("Cleaned link "+str(cleaned_link), RNS.LOG_DEBUG)
+
+ def fail_message(self, lxmessage):
+ RNS.log(str(lxmessage)+" failed to send", RNS.LOG_DEBUG)
+
+ self.pending_outbound.remove(lxmessage)
+ self.failed_outbound.append(lxmessage)
+
+ lxmessage.state = LXMessage.FAILED
+ if lxmessage.failed_callback != None:
+ lxmessage.failed_callback(lxmessage)
+
+ def process_outbound(self, sender = None):
+ if self.processing_outbound:
+ return
+
+ for lxmessage in self.pending_outbound:
+ if lxmessage.state == LXMessage.DELIVERED:
+ RNS.log("Delivery has occurred for "+str(lxmessage)+", removing from outbound queue", RNS.LOG_DEBUG)
+ self.pending_outbound.remove(lxmessage)
+ else:
+ RNS.log("Starting outbound processing for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
+ # Outbound handling for opportunistic messages
+ if lxmessage.method == LXMessage.OPPORTUNISTIC:
+ if lxmessage.delivery_attempts <= LXMRouter.MAX_DELIVERY_ATTEMPTS:
+ if not hasattr(lxmessage, "next_delivery_attempt") or time.time() > lxmessage.next_delivery_attempt:
+ lxmessage.delivery_attempts += 1
+ lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT
+ RNS.log("Opportunistic delivery attempt "+str(lxmessage.delivery_attempts)+" for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
+ lxmessage.send()
+ else:
+ RNS.log("Max delivery attempts reached for oppertunistic "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
+ self.fail_message(lxmessage)
+
+ # Outbound handling for messages transferred
+ # over a direct link to the final recipient
+ elif lxmessage.method == LXMessage.DIRECT:
+ if lxmessage.delivery_attempts <= LXMRouter.MAX_DELIVERY_ATTEMPTS:
+ delivery_destination_hash = lxmessage.get_destination().hash
+
+ if delivery_destination_hash in self.direct_links:
+ # A link already exists, so we'll try to use it
+ # to deliver the message
+ direct_link = self.direct_links[delivery_destination_hash]
+ if direct_link.status == RNS.Link.ACTIVE:
+ if lxmessage.state != LXMessage.SENDING:
+ RNS.log("Starting transfer of "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
+ lxmessage.set_delivery_destination(direct_link)
+ lxmessage.send()
+ else:
+ if lxmessage.representation == LXMessage.RESOURCE:
+ RNS.log("The transfer of "+str(lxmessage)+" is in progress ("+str(round(lxmessage.progress*100, 1))+"%)", RNS.LOG_DEBUG)
+ else:
+ RNS.log("Waiting for proof for "+str(lxmessage)+" sent as link packet", RNS.LOG_DEBUG)
+ elif direct_link.status == RNS.Link.CLOSED:
+ RNS.log("The link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" was closed", RNS.LOG_DEBUG)
+ lxmessage.set_delivery_destination(None)
+ self.direct_links.pop(delivery_destination_hash)
+ lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT
+ else:
+ # Simply wait for the link to become
+ # active or close
+ RNS.log("The link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" is pending, waiting for link to become active", RNS.LOG_DEBUG)
+ else:
+ # No link exists, so we'll try to establish one, but
+ # only if we've never tried before, or the retry wait
+ # period has elapsed.
+ if not hasattr(lxmessage, "next_delivery_attempt") or time.time() > lxmessage.next_delivery_attempt:
+ lxmessage.delivery_attempts += 1
+ lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT
+
+ if lxmessage.delivery_attempts < LXMRouter.MAX_DELIVERY_ATTEMPTS:
+ if RNS.Transport.has_path(lxmessage.get_destination().hash):
+ RNS.log("Establishing link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" for delivery attempt "+str(lxmessage.delivery_attempts)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
+ delivery_link = RNS.Link(lxmessage.get_destination())
+ delivery_link.set_link_established_callback(self.process_outbound)
+ self.direct_links[delivery_destination_hash] = delivery_link
+ else:
+ RNS.log("No path known for delivery attempt "+str(lxmessage.delivery_attempts)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+". Requesting path...", RNS.LOG_DEBUG)
+ RNS.Transport.request_path(lxmessage.get_destination().hash)
+ lxmessage.next_delivery_attempt = time.time() + LXMRouter.PATH_REQUEST_WAIT
+ else:
+ RNS.log("Max delivery attempts reached for direct "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
+ self.fail_message(lxmessage)
+
+ # Outbound handling for messages transported via
+ # propagation to a LXMF router network.
+ elif lxmessage.method == LXMessage.PROPAGATED:
+ RNS.log("Attempting propagated delivery for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
+ raise NotImplementedError("LXMF propagation is not implemented yet")
\ No newline at end of file
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────